package main

import (
	"database/sql"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"strings"
	"time"

	_ "github.com/go-sql-driver/mysql"
)

// 配置结构体
type Config struct {
	MySQL struct {
		Host     string `json:"host"`
		Port     int    `json:"port"`
		User     string `json:"user"`
		Password string `json:"password"`
		Database string `json:"database"`
		Table    string `json:"table"`
	} `json:"mysql"`

	Doris struct {
		Host     string `json:"host"`
		Port     int    `json:"port"`
		User     string `json:"user"`
		Password string `json:"password"`
		Database string `json:"database"`
		Table    string `json:"table"`
	} `json:"doris"`

	Sync struct {
		BatchSize         int    `json:"batch_size"`
		Incremental       bool   `json:"incremental"`
		IncrementalColumn string `json:"incremental_column"`
		LastSyncTime      string `json:"last_sync_time"`
	} `json:"sync"`
}

// 主函数
func main() {
	// 加载配置（实际项目中可以从配置文件读取）
	config := Config{
		MySQL: struct {
			Host     string `json:"host"`
			Port     int    `json:"port"`
			User     string `json:"user"`
			Password string `json:"password"`
			Database string `json:"database"`
			Table    string `json:"table"`
		}{
			Host:     "rm-uf6v4cfq4a000sr3w.mysql.rds.aliyuncs.com",
			Port:     3306,
			User:     "crm_sit_test",
			Password: "14Ckz529dz1i1gDI",
			Database: "mts_sit",
			Table:    "tasks_scheduler",
		},
		Doris: struct {
			Host     string `json:"host"`
			Port     int    `json:"port"`
			User     string `json:"user"`
			Password string `json:"password"`
			Database string `json:"database"`
			Table    string `json:"table"`
		}{
			Host:     "172.22.76.243",
			Port:     9030,
			User:     "crm_test",
			Password: "crm_test",
			Database: "crm_test1",
			Table:    "tasks",
		},
		Sync: struct {
			BatchSize         int    `json:"batch_size"`
			Incremental       bool   `json:"incremental"`
			IncrementalColumn string `json:"incremental_column"`
			LastSyncTime      string `json:"last_sync_time"`
		}{
			BatchSize:         1000,
			Incremental:       true,
			IncrementalColumn: "update_time",
			LastSyncTime:      "2023-01-01 00:00:00",
		},
	}

	// 执行同步
	err := syncMySQLToDoris(config)
	if err != nil {
		log.Fatalf("同步失败: %v", err)
	}
	log.Println("同步完成")
}

// 同步函数
func syncMySQLToDoris(config Config) error {
	// 1. 连接MySQL
	mysqlDB, err := connectMySQL(config.MySQL)
	if err != nil {
		return fmt.Errorf("连接MySQL失败: %v", err)
	}
	defer mysqlDB.Close()

	// 2. 构建查询语句
	query := buildMySQLQuery(config)
	log.Printf("执行查询: %s", query)

	// 3. 执行查询
	rows, err := mysqlDB.Query(query)
	if err != nil {
		return fmt.Errorf("MySQL查询失败: %v", err)
	}
	defer rows.Close()

	// 4. 获取列信息
	columns, err := rows.Columns()
	if err != nil {
		return fmt.Errorf("获取列信息失败: %v", err)
	}

	// 5. 准备Doris写入
	dorisClient := NewDorisClient(config.Doris)

	// 6. 批量读取和处理数据
	batch := make([]map[string]interface{}, 0, config.Sync.BatchSize)
	count := 0

	for rows.Next() {
		// 创建值的容器
		values := make([]interface{}, len(columns))
		valuePtrs := make([]interface{}, len(columns))
		for i := range columns {
			valuePtrs[i] = &values[i]
		}

		// 扫描行数据
		if err := rows.Scan(valuePtrs...); err != nil {
			return fmt.Errorf("扫描行数据失败: %v", err)
		}

		// 构建行map
		rowData := make(map[string]interface{})
		for i, col := range columns {
			val := values[i]
			b, ok := val.([]byte)
			if ok {
				rowData[col] = string(b)
			} else {
				rowData[col] = val
			}
		}

		// 添加到批次
		batch = append(batch, rowData)
		count++

		// 达到批次大小时写入Doris
		if len(batch) >= config.Sync.BatchSize {
			if err := dorisClient.BatchWrite(config.Doris.Database, config.Doris.Table, batch); err != nil {
				return fmt.Errorf("写入Doris失败: %v", err)
			}
			batch = batch[:0] // 清空批次
			log.Printf("已处理 %d 条记录", count)
		}
	}

	// 写入剩余数据
	if len(batch) > 0 {
		if err := dorisClient.BatchWrite(config.Doris.Database, config.Doris.Table, batch); err != nil {
			return fmt.Errorf("写入Doris失败: %v", err)
		}
		log.Printf("已处理 %d 条记录", count)
	}

	// 检查是否有行错误
	if err = rows.Err(); err != nil {
		return fmt.Errorf("行处理错误: %v", err)
	}

	return nil
}

// 连接MySQL
func connectMySQL(cfg struct {
	Host     string `json:"host"`
	Port     int    `json:"port"`
	User     string `json:"user"`
	Password string `json:"password"`
	Database string `json:"database"`
	Table    string `json:"table"`
}) (*sql.DB, error) {
	dsn := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?charset=utf8mb4&parseTime=True&loc=Local",
		cfg.User, cfg.Password, cfg.Host, cfg.Port, cfg.Database)
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		return nil, err
	}
	if err = db.Ping(); err != nil {
		return nil, err
	}
	return db, nil
}

// 构建MySQL查询
func buildMySQLQuery(config Config) string {
	tableName := config.MySQL.Table
	columns := "*" // 实际项目中可以指定列

	if config.Sync.Incremental {
		return fmt.Sprintf("SELECT %s FROM %s WHERE %s > '%s' ORDER BY %s ASC",
			columns, tableName, config.Sync.IncrementalColumn, config.Sync.LastSyncTime, config.Sync.IncrementalColumn)
	}
	return fmt.Sprintf("SELECT %s FROM %s", columns, tableName)
}

// Doris客户端
type DorisClient struct {
	Host     string
	Port     int
	User     string
	Password string
	Client   *http.Client
}

func NewDorisClient(cfg struct {
	Host     string `json:"host"`
	Port     int    `json:"port"`
	User     string `json:"user"`
	Password string `json:"password"`
	Database string `json:"database"`
	Table    string `json:"table"`
}) *DorisClient {
	return &DorisClient{
		Host:     cfg.Host,
		Port:     cfg.Port,
		User:     cfg.User,
		Password: cfg.Password,
		Client:   &http.Client{Timeout: 30 * time.Second},
	}
}

// 批量写入Doris
func (c *DorisClient) BatchWrite(database, table string, data []map[string]interface{}) error {
	if len(data) == 0 {
		return nil
	}

	// 构建Stream Load请求
	url := fmt.Sprintf("http://%s:%d/api/%s/%s/_stream_load",
		c.Host, c.Port, database, table)

	// 将数据转换为JSON格式
	var builder strings.Builder
	for _, row := range data {
		jsonData, err := json.Marshal(row)
		if err != nil {
			return fmt.Errorf("JSON编码失败: %v", err)
		}
		builder.Write(jsonData)
		builder.WriteByte('\n')
	}
	payload := strings.NewReader(builder.String())

	// 创建请求
	req, err := http.NewRequest("PUT", url, payload)
	if err != nil {
		return fmt.Errorf("创建请求失败: %v", err)
	}

	// 设置认证头
	req.SetBasicAuth(c.User, c.Password)
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Expect", "100-continue")
	req.Header.Set("format", "json")
	req.Header.Set("strip_outer_array", "true")

	// 发送请求
	resp, err := c.Client.Do(req)
	if err != nil {
		return fmt.Errorf("请求失败: %v", err)
	}
	defer resp.Body.Close()

	// 检查响应
	if resp.StatusCode != http.StatusOK {
		var result map[string]interface{}
		if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
			return fmt.Errorf("响应解析失败: %v", err)
		}
		return fmt.Errorf("Doris写入失败: %v", result)
	}

	return nil
}
